#pragma once
#include "../comm/Log.h"
#include "../comm/util.h"
#include "../comm/msg.pb.h"
#include <unordered_map>
#include <vector>
#include <functional>
#include <memory>
#include <mutex>

namespace ns_consume
{
    using namespace ns_log;
    using namespace ns_util;
    using namespace ns_proto;

    using ConsumerCallback = function<void(
        const std::string &, const ns_proto::Attributes *, const std::string &)>;
        
    // 消费者描述结构
    struct Consumer
    {
        using ptr = shared_ptr<Consumer>;

        string _qname;              // 订阅队列名称
        string _tag;                // 消费者标识
        bool _auto_ack;             // 自动应答标识
        ConsumerCallback _callback; // 消费者的消息回调函数

        Consumer() {}

        Consumer(const string &qname, const string &tag,
                 const bool &auto_ack, const ConsumerCallback &cb)
            : _tag(tag), _qname(qname), _auto_ack(auto_ack), _callback(cb) {}
    };

    // 队列为单元的消费者管理
    class QueueConsumer
    {
    public:
        using ptr = shared_ptr<QueueConsumer>;

        QueueConsumer(const string &qname)
            : _qname(qname), _seq(0)
        {
        }

        // 创建、删除消费者
        Consumer::ptr insert_consumer(const string &qname, const string &tag,
                                      const bool &auto_ack, const ConsumerCallback &cb)
        {
            unique_lock<mutex> lck(_mtx);

            for (auto &consumer_ptr : _consumers)
            {
                if (consumer_ptr->_tag == tag)
                {
                    LOG(WARNING) << "已经存在相同的的消费者名称" << endl;
                    return Consumer::ptr();
                }
            }
            auto new_consumer = make_shared<Consumer>(tag, qname, auto_ack, cb);
            _consumers.push_back(new_consumer);

            return new_consumer;
        }
        void delete_consumer(const string &tag)
        {
            unique_lock<mutex> lck(_mtx);
            // vector<Consumer::ptr>::iterator
            for (auto it = _consumers.begin(); it < _consumers.end(); it++)
            {
                if ((*it)->_tag == tag)
                {
                    _consumers.erase(it);
                    return;
                }
            }
        }

        // 轮转选择一个消费者进行处理数据
        Consumer::ptr select_consumer()
        {
            unique_lock<mutex> lck(_mtx);
            if (_consumers.empty())
            {
                LOG(WARNING) << "当前不存在任何消费者" << endl;
                return Consumer::ptr();
            }
            int index = _seq % _consumers.size();
            _seq++;
            return _consumers[index];
        }

        // --------------------------------------------测试
        size_t size()
        {
            unique_lock<mutex> lck(_mtx);
            return _consumers.size();
        }
        bool empty()
        {
            unique_lock<mutex> lck(_mtx);
            return _consumers.empty();
        }
        bool exist(const string &tag)
        {
            unique_lock<mutex> lck(_mtx);
            for (auto &consumer_ptr : _consumers)
            {
                if (consumer_ptr->_tag == tag)
                {
                    return true;
                }
            }
            return false;
        }
        void clear()
        {
            unique_lock<mutex> lck(_mtx);
            _consumers.clear();
            _seq = 0;
        }

    private:
        mutex _mtx;
        string _qname;
        uint16_t _seq;                    // 消费者序号(用于轮转消费者)
        vector<Consumer::ptr> _consumers; // 消费者管理
    };

    // 所有队列消费者管理
    class ConsumerManage
    {
    public:
        ConsumerManage() {}

        // 初始化、销毁队列消费者
        void init_queue_consumer(const string &qname)
        {
            unique_lock<mutex> lck(_mtx);
            if (_qconsumers.count(qname))
            {
                LOG(WARNING) << "已存在该名称的消费者队列" << endl;
                return;
            }
            QueueConsumer::ptr new_qconsumer = make_shared<QueueConsumer>(qname);
            _qconsumers.insert(make_pair(qname, new_qconsumer));
        }
        void destroy_queue_consumer(const string &qname)
        {
            unique_lock<mutex> lck(_mtx);
            if (!_qconsumers.count(qname))
            {
                LOG(WARNING) << "不存在该名称的消费者队列" << endl;
                return;
            }
            _qconsumers.erase(qname);
        }

        // 向指定队列插入、删除消费者
        Consumer::ptr insert_queue_consumer(const string &qname, const string &tag,
                                            const bool &auto_ack, const ConsumerCallback &cb)
        {
            {
                unique_lock<mutex> lck(_mtx);
                if (!_qconsumers.count(qname))
                {
                    LOG(WARNING) << "不存在该名称的消费者队列" << endl;
                    return Consumer::ptr();
                }
            }
            return _qconsumers[qname]->insert_consumer(tag, qname, auto_ack, cb);
        }

        void delete_queue_consumer(const string &qname, const string &tag)
        {
            {
                unique_lock<mutex> lck(_mtx);
                if (!_qconsumers.count(qname))
                {
                    LOG(WARNING) << "不存在该名称的消费者队列" << endl;
                    return;
                }
            }
            _qconsumers[qname]->delete_consumer(tag);
        }

        // 轮转选择一个队列消费者
        Consumer::ptr select_queue_consumer(const string &qname)
        {
            {
                unique_lock<mutex> lck(_mtx);
                if (!_qconsumers.count(qname))
                {
                    LOG(WARNING) << "不存在该名称的消费者队列" << endl;
                    return Consumer::ptr();
                }
            }
            return _qconsumers[qname]->select_consumer();
        }

        // --------------------------------------------测试
        size_t size(const string &qname)
        {
            {
                unique_lock<mutex> lck(_mtx);
                if (!_qconsumers.count(qname))
                {
                    LOG(WARNING) << "不存在该名称的消费者队列" << endl;
                    return 0;
                }
            }
            return _qconsumers[qname]->size();
        }
        bool empty(const string &qname)
        {
            {
                unique_lock<mutex> lck(_mtx);
                if (!_qconsumers.count(qname))
                {
                    LOG(WARNING) << "不存在该名称的消费者队列" << endl;
                    return false;
                }
            }
            return _qconsumers.empty();
        }
        bool exist(const string &qname, const string &tag)
        {
            {
                unique_lock<mutex> lck(_mtx);
                if (!_qconsumers.count(qname))
                {
                    LOG(WARNING) << "不存在该名称的消费者队列" << endl;
                    return false;
                }
            }
            return _qconsumers[qname]->exist(tag);
        }
        void clear()
        {
            unique_lock<mutex> lck(_mtx);
            _qconsumers.clear();
        }

    private:
        mutex _mtx; // 保证_qconsumers操作的安全性
        unordered_map<string, shared_ptr<QueueConsumer>> _qconsumers;
    };
}